package com.atguigu.member.service

import com.alibaba.fastjson.JSONObject
import com.atguigu.constan.Constan
import com.atguigu.util.JsonUtil
import org.apache.spark.sql.{SaveMode, SparkSession}

/**
 * description ：etl ods 数据原始层
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/3/6 15:05
 * modified By ：
 * version:    : 1.0
 */
object EtlOdsDataService {

  /**
   * vip基础数据
   *
   * @Author 剧情再美终是戏
   * @Date 2020/3/6 15:14
   * @param spark
   * @return java.lang.Object
   * @Version 1.0
   **/
  def etlPcenterMemViplevel(spark: SparkSession) = {
    import spark.implicits._
    // 读取源数据
    val ssc = spark.sparkContext
    val source = ssc.textFile(Constan.ODS_FILE_PATH + "/pcenterMemViplevel.log")

    // etl 并将数据写入 `dwd`.`dwd_pcentermempaymoney`
    source
      .filter(line => JsonUtil.parseObject(line).isInstanceOf[JSONObject])
      .mapPartitions(it => {
        it.map(line => {
          val jsonObj = JsonUtil.parseObject(line)
          val vip_id = jsonObj.getIntValue("vip_id")
          val vip_level = jsonObj.getString("vip_level")
          val start_time = jsonObj.getString("start_time")
          val end_time = jsonObj.getString("end_time")
          val last_modify_time = jsonObj.getString("last_modify_time")
          val max_free = jsonObj.getString("max_free")
          val min_free = jsonObj.getString("min_free")
          val next_level = jsonObj.getString("next_level")
          val operator = jsonObj.getString("operator")
          val dn = jsonObj.getString("dn")
          (vip_id, vip_level, start_time, end_time, last_modify_time, max_free, min_free, next_level, operator, dn)
        })
      })
      .toDF().coalesce(1).write.mode(SaveMode.Overwrite).insertInto("dwd.dwd_vip_level")
  }

  /**
   * 用户支付情况记录
   *
   * @Author 剧情再美终是戏
   * @Date 2020/3/6 15:13
   * @param spark
   * @return java.lang.Object
   * @Version 1.0
   **/
  def etlPcentermempayMoney(spark: SparkSession) = {
    import spark.implicits._
    // 读取源数据
    val ssc = spark.sparkContext
    val source = ssc.textFile(Constan.ODS_FILE_PATH + "/pcentermempaymoney.log")

    // etl 并将数据写入 `dwd`.`dwd_pcentermempaymoney`
    source
      .filter(line => JsonUtil.parseObject(line).isInstanceOf[JSONObject])
      .mapPartitions(it => {
        it.map(line => {
          val jsonObj = JsonUtil.parseObject(line)
          val uid = jsonObj.getIntValue("uid")
          val paymoney = jsonObj.getString("paymoney")
          val siteid = jsonObj.getIntValue("siteid")
          val vip_id = jsonObj.getIntValue("vip_id")
          val dt = jsonObj.getString("dt")
          val dn = jsonObj.getString("dn")
          (uid, paymoney, siteid, vip_id, dt, dn)
        })
      })
      .toDF().coalesce(1).write.mode(SaveMode.Overwrite).insertInto("dwd.dwd_pcentermempaymoney")
  }

  /**
   * 用户数据
   *
   * @Author 剧情再美终是戏
   * @Date 2020/3/6 15:13
   * @param spark
   * @return java.lang.Object
   * @Version 1.0
   **/
  def etlMember(spark: SparkSession) = {
    import spark.implicits._
    // 读取源数据
    val ssc = spark.sparkContext
    val source = ssc.textFile(Constan.ODS_FILE_PATH + "/member.log")

    // etl && 将结果写入 `dwd`.`dwd_member`
    source
      .filter(line => JsonUtil.parseObject(line).isInstanceOf[JSONObject])
      .mapPartitions(it => {
        it.map(line => {
          // 清洗规则 用户名：王XX   手机号：137*****789  密码直接替换成******
          val jsonObj = JsonUtil.parseObject(line)
          val uid = jsonObj.getIntValue("uid")
          val ad_id = jsonObj.getIntValue("ad_id")
          val birthday = jsonObj.getString("birthday")
          val email = jsonObj.getString("email")
          val fullname = jsonObj.getString("fullname").charAt(0).toString + "XX"
          val iconurl = jsonObj.getString("iconurl")
          val lastlogin = jsonObj.getString("lastlogin")
          val mailaddr = jsonObj.getString("mailaddr")
          val memberlevel = jsonObj.getString("memberlevel")
          val password = "******"
          val paymoney = jsonObj.getString("paymoney")
          val phone = jsonObj.getString("phone").replaceAll("(\\{3})(\\{4})(\\{4})", "$1****$3")
          val qq = jsonObj.getString("qq")
          val register = jsonObj.getString("register")
          val regupdatetime = jsonObj.getString("regupdatetime")
          val unitname = jsonObj.getString("unitname")
          val userip = jsonObj.getString("userip")
          val zipcode = jsonObj.getString("zipcode")
          val dt = jsonObj.getString("dt")
          val dn = jsonObj.getString("dn")
          (uid, ad_id, birthday, email, fullname, iconurl, lastlogin, mailaddr, memberlevel, password, paymoney,
            phone, qq, register, regupdatetime, unitname, userip, zipcode, dt, dn)
        })
      })
      .toDF().coalesce(2).write.mode(SaveMode.Overwrite).insertInto("dwd.dwd_member")
  }

  /**
   * 用户注册数据
   *
   * @Author 剧情再美终是戏
   * @Date 2020/3/6 15:13
   * @param spark
   * @return java.lang.Object
   * @Version 1.0
   **/
  def etlMemberRegtype(spark: SparkSession) = {
    import spark.implicits._
    // 读取源数据
    val ssc = spark.sparkContext
    val source = ssc.textFile(Constan.ODS_FILE_PATH + "/memberRegtype.log")

    // etl && 将数据写入 `dwd`.`dwd_member_regtype` 表中
    source
      .filter(line => JsonUtil.parseObject(line).isInstanceOf[JSONObject])
      .mapPartitions(it => {
        it.map(line => {
          val jsonObj = JsonUtil.parseObject(line)
          val uid = jsonObj.getIntValue("uid")
          val appkey = jsonObj.getString("appkey")
          val appregurl = jsonObj.getString("appregurl")
          val bdp_uuid = jsonObj.getString("bdp_uuid")
          val createtime = jsonObj.getString("createtime")
          val domain = jsonObj.getString("domain")
          val isranreg = jsonObj.getString("isranreg")
          val regsource = jsonObj.getString("regsource")
          val regsourcename = regsource match {
            case "1" => "PC"
            case "2" => "Mobile"
            case "3" => "App"
            case "4" => "WeChat"
            case _ => "other"
          }
          val websiteid = jsonObj.getIntValue("websiteid")
          val dt = jsonObj.getString("dt")
          val dn = jsonObj.getString("dn")
          (uid, appkey, appregurl, bdp_uuid, createtime, domain, isranreg, regsource, regsourcename, websiteid, dt, dn)
        })
      })
      .toDF().coalesce(1).write.mode(SaveMode.Overwrite).insertInto("dwd.dwd_member_regtype")
  }

  /**
   * 基础网站表数据
   *
   * @Author 剧情再美终是戏
   * @Date 2020/3/6 15:13
   * @param spark
   * @return java.lang.Object
   * @Version 1.0
   **/
  def etlBasweWebsite(spark: SparkSession) = {
    import spark.implicits._
    // 读取 ods baswewebsite.log 数据
    val ssc = spark.sparkContext
    val source = ssc.textFile(Constan.ODS_FILE_PATH + "/baswewebsite.log")

    // etl 并将数据写入 dwd dwd_base_website
    source
      .filter(line => JsonUtil.parseObject(line).isInstanceOf[JSONObject])
      .mapPartitions(it => {
        it.map(line => {
          val jsonObj = JsonUtil.parseObject(line)
          val siteid = jsonObj.getIntValue("siteid")
          val sitename = jsonObj.getString("sitename")
          val siteurl = jsonObj.getString("siteurl")
          val delete = jsonObj.getIntValue("delete")
          val createtime = jsonObj.getString("createtime")
          val creator = jsonObj.getString("creator")
          val dn = jsonObj.getString("dn")
          (siteid, sitename, siteurl, delete, createtime, creator, dn)
        })
      })
      .toDF().coalesce(2).write.mode(SaveMode.Overwrite).insertInto("dwd.dwd_base_website")
  }

  /**
   * 基础广告表数据
   *
   * @Author 剧情再美终是戏
   * @Date 2020/3/6 15:12
   * @param spark
   * @return java.lang.Object
   * @Version 1.0
   **/
  def etlBaseadLog(spark: SparkSession) = {
    //隐式转换
    import spark.implicits._

    // 读取 ods baseadLog.log 数据
    val ssc = spark.sparkContext
    val source = ssc.textFile(Constan.ODS_FILE_PATH + "/baseadlog.log")

    // etl && 将数据写入 dwd baseadLog
    source
      .filter(line => JsonUtil.parseObject(line).isInstanceOf[JSONObject])
      .mapPartitions(it => {
        it.map(line => {
          val jsonObj = JsonUtil.parseObject(line)
          val adid = jsonObj.getIntValue("adid")
          val adname = jsonObj.getString("adname")
          val dn = jsonObj.getString("dn")
          (adid, adname, dn)
        })
      })
      .toDF().coalesce(1).write.mode(SaveMode.Overwrite).insertInto("dwd.dwd_base_ad")
  }
}
